package com.yongche

import java.io.File

import com.yongche.kafka.KafkaInfo
import com.yongche.utils.ArgumentParser
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Minutes, StreamingContext}

import scala.io.Source

/**
  * Created by kuan on 2016/10/9.
  */
object App {
  def main(args: Array[String]): Unit = {
    //如果参数的长度不为1，那么直接报错
    if (args.length != 1) {
      println("需要1个参数，请输入配置文件的绝对路径")
      System.exit(-1)
    }

    //转换过的参数
    val parsedArgs = ArgumentParser
      .parse(Source.fromFile(new File(args(0))).mkString)

    //各个参数
    val (Some(topic), Some(brokers), Some(zkHost), Some(slideDuration), Some(timeField)) = parsedArgs

    //spark配置项
    val sparkConf = new SparkConf().setAppName(s"${topic}_ETL")
    //创建流式环境
    val ssc: StreamingContext = new StreamingContext(new SparkContext(sparkConf), Minutes(slideDuration.toLong))
    //设置流计算的持久化目录，必须设置，但是运行环境的恢复由程序自行控制
    ssc.checkpoint(s"/tmp/etl/${topic}/checkpoint")
    ssc.sparkContext.getConf.set("spark.streaming.kafka.maxRatePerPartition", (1000 * 1000).toString)





    //用到的spark sql
    implicit val sqlContext: SQLContext = new SQLContext(ssc.sparkContext)
    import sqlContext.implicits._

    //kafka信息
    val kafkaInfo = KafkaInfo(topic)
    //kafka输入流
    val streams = kafkaInfo.streaming(ssc)(brokers, zkHost)

    //kafka工作流
    val ds = streams.mapPartitions(lineTransfer(timeField)).window(Minutes(slideDuration), Minutes(slideDuration))
    ds.persist(StorageLevel.MEMORY_AND_DISK)
    //对于每个RDD进行操作
    ds.foreachRDD { (rdd, timestamp) =>
      //获得所有的时间的列表
      val allTimeStr = rdd.map(x => x._1).distinct().collect()
      //写入到hdfs的方法
      val write2HDFS = kafkaInfo.write2HDFS(_.toDF())(rdd, timestamp)_
      //对每个时间的数据保存到不同的路径下
      allTimeStr.foreach(write2HDFS)
    }
    //程序正式启动
    ssc.start()
    ssc.awaitTermination()
  }

  /**
    * 根据时间进行划分数据
    *
    * @param timeField 时间的字段索引和转换方法
    * @param it        数据
    * @return 转换完成的数据
    */
  def lineTransfer(timeField: (Int, (String) => String))(it: Iterator[String]): Iterator[(String, String)] = {
    //日志时间戳
    val (index, transferFun) = timeField
    it.map { line =>
      try {
        //时间
        val dtStr = line.split("\\s")(index)
        //转换成(时间,日志内容)
        (transferFun(dtStr), line)
      } catch {
        //如果抛出异常的话，则第一个字段为null，或许可以对其进行操作
        case ex: Exception => (null, line)
      }
    }
  }
}